package com.starrocks.controller;

import com.starrocks.log.LogCollector;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.function.Consumer;

/**
 * 实时日志控制器
 * 提供 SSE 接口推送后端日志到前端
 */
@Slf4j
@RestController
@RequestMapping("/api/logs")
public class LogController {
    
    @Autowired
    private LogCollector logCollector;
    
    /**
     * 获取历史日志（最近的1000条）
     */
    @GetMapping("/history")
    public ConcurrentLinkedQueue<LogCollector.LogMessage> getHistoryLogs() {
        return logCollector.getHistoryLogs();
    }
    
    /**
     * SSE 实时日志流
     * 客户端通过 EventSource 连接此接口，接收实时日志
     */
    @SuppressWarnings("unchecked")
    @GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public SseEmitter streamLogs(@RequestParam(required = false, defaultValue = "INFO") String level) {
        // 创建 SSE emitter（超时时间：30分钟）
        SseEmitter emitter = new SseEmitter(30 * 60 * 1000L);
        
        log.info("新的日志流连接建立，日志级别过滤：{}", level);
        
        // 使用数组包装consumer，使其在lambda中可变
        final Consumer<LogCollector.LogMessage>[] consumerHolder = new Consumer[1];
        
        // 订阅日志
        Consumer<LogCollector.LogMessage> consumer = logMessage -> {
            try {
                // 根据日志级别过滤
                if (shouldSendLog(logMessage.getLevel(), level)) {
                    // 发送日志到前端
                    emitter.send(SseEmitter.event()
                        .name("log")
                        .data(logMessage.toJson()));
                }
            } catch (IOException e) {
                log.warn("发送日志到客户端失败，移除订阅", e);
                if (consumerHolder[0] != null) {
                    logCollector.unsubscribe(consumerHolder[0]);
                }
                emitter.completeWithError(e);
            }
        };
        
        consumerHolder[0] = consumer;
        logCollector.subscribe(consumer);
        
        // 当连接关闭时，取消订阅
        emitter.onCompletion(() -> {
            log.info("日志流连接关闭（正常完成）");
            if (consumerHolder[0] != null) {
                logCollector.unsubscribe(consumerHolder[0]);
            }
        });
        
        emitter.onTimeout(() -> {
            log.info("日志流连接超时");
            if (consumerHolder[0] != null) {
                logCollector.unsubscribe(consumerHolder[0]);
            }
            emitter.complete();
        });
        
        emitter.onError(e -> {
            log.warn("日志流连接异常", e);
            if (consumerHolder[0] != null) {
                logCollector.unsubscribe(consumerHolder[0]);
            }
        });
        
        // 发送连接成功消息
        try {
            emitter.send(SseEmitter.event()
                .name("connected")
                .data("{\"message\":\"日志流连接成功\",\"subscriberCount\":" + logCollector.getSubscriberCount() + "}"));
        } catch (IOException e) {
            log.error("发送连接成功消息失败", e);
        }
        
        return emitter;
    }
    
    /**
     * 获取当前订阅者数量
     */
    @GetMapping("/subscriber-count")
    public int getSubscriberCount() {
        return logCollector.getSubscriberCount();
    }
    
    /**
     * 判断是否应该发送该日志（根据级别过滤）
     */
    private boolean shouldSendLog(String logLevel, String filterLevel) {
        // 如果过滤级别为 ALL，发送所有日志
        if ("ALL".equalsIgnoreCase(filterLevel)) {
            return true;
        }
        
        // 日志级别优先级：TRACE < DEBUG < INFO < WARN < ERROR
        int logLevelPriority = getLevelPriority(logLevel);
        int filterLevelPriority = getLevelPriority(filterLevel);
        
        return logLevelPriority >= filterLevelPriority;
    }
    
    /**
     * 获取日志级别优先级
     */
    private int getLevelPriority(String level) {
        switch (level.toUpperCase()) {
            case "TRACE":
                return 1;
            case "DEBUG":
                return 2;
            case "INFO":
                return 3;
            case "WARN":
                return 4;
            case "ERROR":
                return 5;
            default:
                return 3; // 默认 INFO
        }
    }
}

